package net.bwie.realtime.jtp.dws.trade.function;

import net.bwie.realtime.jtp.dws.trade.bean.LiveRoomEventBean;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.HashSet;
import java.util.Set;

public class AggregateLiveRoomEventMetrics implements WindowFunction<LiveRoomEventBean, LiveRoomEventBean, String, TimeWindow> {

    private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai");
    private static final DateTimeFormatter FMT_TS =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZONE);
    private static final DateTimeFormatter FMT_MIN =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm").withZone(ZONE);

    // ✅ 工具方法
    private static boolean isBlank(String s) {
        return s == null || s.trim().isEmpty();
    }

    private static boolean validUser(String u) {
        return u != null && !u.isEmpty();
    }


    @Override
    public void apply(String key, TimeWindow window, Iterable<LiveRoomEventBean> elements,
                      Collector<LiveRoomEventBean> out) {

        int dealCount = 0;
        double dealAmount = 0D;
        Set<String> dealUserSet = new HashSet<>();
        Set<String> uvSet = new HashSet<>();
        long durationSec = 0L;
        int newFansCount = 0;
        int newFansTeamCount = 0;
        int commentCount = 0;

        String liveRoomId = null, anchorId = null, eventMinute = null;

        for (LiveRoomEventBean bean : elements) {
            String et = bean.getEventType();
            if (et != null) et = et.trim();

            if (isBlank(liveRoomId) && !isBlank(bean.getLiveRoomId())) liveRoomId = bean.getLiveRoomId();
            if (isBlank(anchorId) && !isBlank(bean.getAnchorId())) anchorId = bean.getAnchorId();
            if (isBlank(eventMinute) && !isBlank(bean.getEventMinute())) eventMinute = bean.getEventMinute();

            // ✅ 成交（按支付为准）
            if ("支付".equals(et)) {
                dealCount++;
                if (validUser(bean.getUserId())) dealUserSet.add(bean.getUserId());
                if (bean.getOrderAmount() != null) dealAmount += bean.getOrderAmount();
            }

            // ✅ UV（任何事件，只要有 userId 就统计一次）
            if (validUser(bean.getUserId())) {
                uvSet.add(bean.getUserId());
            }

            // ✅ 停留时长（任意事件只要有 durationSec 就加）
            if (bean.getDurationSec() != null && bean.getDurationSec() > 0) {
                durationSec += bean.getDurationSec();
            }

            // ✅ 新增粉丝（sum isNew 字段）
            newFansCount += (bean.getIsNew() == null ? 0 : bean.getIsNew());

            // ✅ 新增粉丝团（靠事件类型）
            if ("加入粉丝团".equals(et)) {
                newFansTeamCount++;
            }

            // ✅ 评论（sum isNewComment 字段）
            commentCount += (bean.getIsNewComment() == null ? 0 : bean.getIsNewComment());
        }

        // ✅ 聚合结果
        LiveRoomEventBean result = new LiveRoomEventBean();
        result.setLiveRoomId(liveRoomId);
        result.setAnchorId(anchorId);
        result.setDealCount(dealCount);
        result.setDealUserCount(dealUserSet.size());
        result.setDealAmount(dealAmount);
        result.setUvCount(uvSet.size());
        result.setDurationSec(durationSec);
        result.setNewFansCount(newFansCount);
        result.setNewFansTeamCount(newFansTeamCount);
        result.setCommentCount(commentCount);

        // ✅ 窗口时间
        //kkkkkk
        result.setWindowStartTime(FMT_TS.format(Instant.ofEpochMilli(window.getStart())));
        result.setWindowEndTime(FMT_TS.format(Instant.ofEpochMilli(window.getEnd())));
        if (isBlank(eventMinute)) {
            eventMinute = FMT_MIN.format(Instant.ofEpochMilli(window.getStart()));
        }
        result.setEventMinute(eventMinute);

        result.setTs(System.currentTimeMillis());

        out.collect(result);
    }

}
